package cn.mesmile.flink.window;

import cn.mesmile.flink.jdkstream.VideoOrder;
import cn.mesmile.flink.source.VideoOrderSourceV2;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Date;

/**
 * @author zb
 * @date 2021/8/23 22:46
 * @Description  增量聚合函数
 */
public class FlinkAggWindowsApp {

    /**
     * - 增量聚合函数
     *   aggregate(agg函数,WindowFunction(){  })
     *   - 【窗口保存临时数据，每进入一个新数据，会与中间数据累加】，生成新的中间数据，再保存到窗口中
     *   - 常见的增量聚合函数有 reduceFunction、aggregateFunction
     *   - min、max、sum 都是简单的聚合操作，不需要自定义规则
     *   AggregateFunction<IN, ACC, OUT>
     *   IN是输入类型，ACC是中间聚合状态类型，OUT是输出类型，是聚合统计当前窗口的数据
     */
    public static void main(String[] args) throws Exception {

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 设置成 单个线程运行 方便查询演示效果
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStreamSource<VideoOrder> ds = env.addSource(new VideoOrderSourceV2());

        KeyedStream<VideoOrder, String> keyByDb = ds.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder videoOrder) throws Exception {
                return videoOrder.getTitle();
            }
        });

        /*
        AggregateFunction<IN, ACC, OUT>
        IN是输入类型，ACC是中间聚合状态类型，OUT是输出类型，是聚合统计当前窗口的数据
         */
        SingleOutputStreamOperator<VideoOrder> aggregate = keyByDb.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<VideoOrder, VideoOrder, VideoOrder>() {
                    // 初始化累加器
                    @Override
                    public VideoOrder createAccumulator() {
                        VideoOrder videoOrder = new VideoOrder();
                        return videoOrder;
                    }
                    // 聚合方式
                    @Override
                    public VideoOrder add(VideoOrder value, VideoOrder accumulator) {
                        accumulator.setMoney(value.getMoney()+ accumulator.getMoney());
                        accumulator.setTitle(value.getTitle());
                        if (accumulator.getTitle() == null){
                            accumulator.setTitle(value.getTitle());
                        }
                        if (accumulator.getCreateTime() == null){
                            accumulator.setCreateTime(value.getCreateTime());
                        }
                        return accumulator;
                    }
                    // 获取结果
                    @Override
                    public VideoOrder getResult(VideoOrder accumulator) {
                        return accumulator;
                    }
                    // 一般不用
                    @Override
                    public VideoOrder merge(VideoOrder a, VideoOrder b) {
                        VideoOrder videoOrder = new VideoOrder();
                        videoOrder.setMoney(a.getMoney() + b.getMoney());
                        videoOrder.setTitle(a.getTitle());
                        return videoOrder;
//                        return null;
                    }
                });

        aggregate.print("result:");

        env.execute("count windows job");
    }
}
